package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * @author jiwei
 * @description 模拟甲烷浓度数据写入Kafka
 * @date 2023/6/12 9:02
 */
public class Datagen2Kafka_MethaneConcentration {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //设置时区
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    device_id       string,\n" +     //设备编号
                "    device_name      string,\n" +    //设备名称
                "    concentration   double,\n" +    //甲烷浓度值
                "    `timestamp`   timestamp(3)\n" +       //时间戳
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '50000000', \n" +
                "  'rows-per-second' = '10000', \n" +
                "  'fields.device_name.length' = '5', \n" +
                "  'fields.device_id.length' = '5', \n" +
                "  'fields.concentration.min' = '10', \n" +
                "  'fields.concentration.max' = '200'\n" +
                ")");

        //Sink
        tableEnv.executeSql("CREATE TABLE sinkTable (" +
                "    device_id       string,\n" +     //设备编号
                "    device_name      string,\n" +    //设备名称
                "    concentration   double,\n" +    //甲烷浓度值
                "    `timestamp`   timestamp(3)\n" +       //时间戳
                "   ) " +
                "   WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'std.object.methane',\n" +
                "  'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
                "  'properties.allow.auto.create.topics' = 'true',\n" +
//                "  'sink.parallelism' = '5',\n" +
//                "  'format' = 'csv'\n" +
                "  'format' = 'json'\n" +
//                "  'format' = 'debezium-json'\n" +
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable select * from sourceTable");

    }
}
